package com.lfy.modules.monitor.task;

import com.lfy.modules.monitor.domain.vo.SystemCpuUsingEvent;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.MediaType;
import org.springframework.util.CollectionUtils;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.util.Objects;
import java.util.concurrent.CopyOnWriteArraySet;
/**
 *  sse
 *
 * @author <a href="https://www.lvfeiy.top">lvfeiyang</a>
 * @date 2024/4/25 1:03
 */
@Slf4j
public class SseTaskCache {
    private static final CopyOnWriteArraySet<SseEmitter> EMITTERS = new CopyOnWriteArraySet<>();
    public static final Long TIME_OUT = 1000 * 60 * 60L;

    private static void add(SseEmitter sseEmitter) {
        if (!Objects.isNull(sseEmitter)) {
            EMITTERS.add(sseEmitter);
        }
    }

    private static void remove(SseEmitter sseEmitter) {
        if (!Objects.isNull(sseEmitter)) {
            EMITTERS.remove(sseEmitter);
        }
    }

    public static boolean isEmpty() {
        return EMITTERS.isEmpty();
    }

    public static void sendMsg(SystemCpuUsingEvent event) {
        if (CollectionUtils.isEmpty(EMITTERS)) {
            return;
        }
        for (SseEmitter emitter : EMITTERS) {
            try {
                emitter.send(event, MediaType.APPLICATION_JSON);
            } catch (IOException e) {
                log.warn("发送sse消息失败，连接已中断 {}", event);
            }
        }
    }


    public static SseEmitter createChannel() {
        SseEmitter sseEmitter = new SseEmitter(0L);
        sseEmitter.onError(e -> {
            log.warn("sse 中断连接");
            remove(sseEmitter);
        });
        sseEmitter.onCompletion(() -> {
            remove(sseEmitter);
        });
        sseEmitter.onTimeout(() -> {
            remove(sseEmitter);
        });
        add(sseEmitter);
        return sseEmitter;
    }
}
